package com.dj.ms.support.mq;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.aliyun.openservices.ons.api.*;
import com.aliyun.openservices.ons.api.order.OrderConsumer;
import com.aliyun.openservices.ons.api.order.OrderProducer;
import com.dj.common.exception.BaseRunTimeException;
import com.dj.ms.configure.AliYunMQConfiguration;
import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;

import javax.annotation.Resource;
import java.util.Map;
import java.util.Properties;
import java.util.UUID;

/**
 * <br>
 * <p>
 * - rocket mq支持
 * </p>
 *
 * <br>
 *
 * @author ZJG
 * @version 1.0
 * @date 2021/10/28 上午7:15
 */
@Slf4j
@Component
public class RocketMQHandler {
    @Resource
    private MqProperties mqProperties;
    @Autowired
    private AliYunMQConfiguration aliYunMQConfiguration;

    private Map<String, Admin> mqAdminMap = Maps.newConcurrentMap();
    private static final String MQ_PRODUCER_KEY = "Producer";
    private static final String MQ_CONSUMER_KEY = "Consumer";
    private static final String MQ_ORDER_KEY = "Order";
    private String groupId;
    private Class mq_class;
    private MQParameter.Coordinator coordinator = new MQParameter.Coordinator();

    /**
     * mq 业务处理助手
     * @param groupId                   mq group
     * @param mq_class                  mq生产者或者消费者class
     *        {@link Producer}
     *        {@link OrderProducer}
     *        {@link Consumer}
     *        {@link OrderConsumer}
     * @param coordinator               mq参数协调者，接收发送或消费参数
     * @return
     */
    public synchronized <T> T process(String groupId, Class<T> mq_class, java.util.function.Consumer<MQParameter.Coordinator> coordinator) {
        Assert.isTrue(StrUtil.isNotBlank(groupId), "groupId不能为空");
        Assert.isTrue(ObjectUtil.isNotNull(mq_class), "mq_producer_class不能为空");
        Assert.isTrue(ObjectUtil.isNotNull(coordinator), "coordinator不能为空");
        T t = doLoad(groupId, mq_class);
        this.groupId = groupId;
        this.mq_class = mq_class;
        coordinator.accept(this.coordinator);
        this.process();
        return t;
    }

    /**
     * 加载操作实例
     * @param groupId
     * @param mq_class
     * @param <T>
     * @return
     */
    private <T> T doLoad(String groupId, Class<T> mq_class) {
        if (StrUtil.containsIgnoreCase(mq_class.getName(), MQ_PRODUCER_KEY)) {
            if (StrUtil.containsIgnoreCase(mq_class.getName(), MQ_ORDER_KEY)) {
                return (T) orderProducer(groupId);
            }
            return (T) generalProducer(groupId);
        }

        if (StrUtil.containsIgnoreCase(mq_class.getName(), MQ_ORDER_KEY)) {
            return (T) orderConsumer(groupId);
        }
        return (T) generalConsumer(groupId);
    }


    /**
     * 通用消费者
     * @param groupId
     * @return
     */
    private Consumer generalConsumer(String groupId) {
        if (CollectionUtil.isNotEmpty(mqAdminMap)
                && mqAdminMap.containsKey(groupId)) {
            return (Consumer) mqAdminMap.get(groupId);
        }
        Properties properties = new Properties();
        properties.put(PropertyKeyConst.GROUP_ID, groupId);
        properties.put(PropertyKeyConst.AccessKey, aliYunMQConfiguration.getAccessKey());
        properties.put(PropertyKeyConst.SecretKey, aliYunMQConfiguration.getSecretKey());
        properties.put(PropertyKeyConst.NAMESRV_ADDR, aliYunMQConfiguration.getNamesrv_addr());
        properties.put(PropertyKeyConst.ConsumeTimeout, mqProperties.getConsumeTimeout());
        properties.put(PropertyKeyConst.MaxReconsumeTimes, mqProperties.getMaxReconsumeTimes());
        properties.put(PropertyKeyConst.SuspendTimeMillis, mqProperties.getSuspendTimeMillis());
        Consumer consumer = ONSFactory.createConsumer(properties);
        consumer.start();
        if (ObjectUtil.isNull(consumer)) {
            throw new BaseRunTimeException("获取generalConsumer异常");
        }
        mqAdminMap.put(groupId, consumer);
        return consumer;
    }

    /**
     * 顺序消费者
     * @param groupId
     * @return
     */
    private OrderConsumer orderConsumer(String groupId) {
        if (CollectionUtil.isNotEmpty(mqAdminMap)
                && mqAdminMap.containsKey(groupId)) {
            return (OrderConsumer) mqAdminMap.get(groupId);
        }
        Properties properties = new Properties();
        properties.put(PropertyKeyConst.GROUP_ID, groupId);
        properties.put(PropertyKeyConst.AccessKey, aliYunMQConfiguration.getAccessKey());
        properties.put(PropertyKeyConst.SecretKey, aliYunMQConfiguration.getSecretKey());
        properties.put(PropertyKeyConst.NAMESRV_ADDR, aliYunMQConfiguration.getNamesrv_addr());
        properties.put(PropertyKeyConst.SuspendTimeMillis, mqProperties.getSuspendTimeMillis());
        properties.put(PropertyKeyConst.ConsumeTimeout, mqProperties.getConsumeTimeout());
        properties.put(PropertyKeyConst.MaxReconsumeTimes, mqProperties.getMaxReconsumeTimes());
        OrderConsumer orderedConsumer = ONSFactory.createOrderedConsumer(properties);
        orderedConsumer.start();
        if (ObjectUtil.isNull(orderedConsumer)) {
            throw new BaseRunTimeException("获取OrderConsumer异常");
        }
        mqAdminMap.put(groupId, orderedConsumer);
        return orderedConsumer;
    }

    /**
     * 顺序消息
     * @param groupId
     * @return
     */
    private OrderProducer orderProducer(String groupId) {
        if (CollectionUtil.isNotEmpty(mqAdminMap)
                && mqAdminMap.containsKey(groupId)) {
            return (OrderProducer) mqAdminMap.get(groupId);
        }
        Properties properties = new Properties();
        properties.put(PropertyKeyConst.GROUP_ID, groupId);
        properties.put(PropertyKeyConst.AccessKey, aliYunMQConfiguration.getAccessKey());
        properties.put(PropertyKeyConst.SecretKey, aliYunMQConfiguration.getSecretKey());
        properties.put(PropertyKeyConst.NAMESRV_ADDR, aliYunMQConfiguration.getNamesrv_addr());
        properties.put(PropertyKeyConst.SendMsgTimeoutMillis, mqProperties.getSendMsgTimeout());
        OrderProducer producer = ONSFactory.createOrderProducer(properties);
        producer.start();
        if (ObjectUtil.isNull(producer)) {
            throw new BaseRunTimeException("获取OrderProducer异常");
        }
        mqAdminMap.put(groupId, producer);
        return producer;
    }

    /**
     * 通用消息
     * @param groupId
     * @return
     */
    private Producer generalProducer(String groupId) {
        if (CollectionUtil.isNotEmpty(mqAdminMap)
                && mqAdminMap.containsKey(groupId)) {
            return (Producer) mqAdminMap.get(groupId);
        }
        Properties properties = new Properties();
        properties.put(PropertyKeyConst.GROUP_ID, groupId);
        properties.put(PropertyKeyConst.AccessKey, aliYunMQConfiguration.getAccessKey());
        properties.put(PropertyKeyConst.SecretKey, aliYunMQConfiguration.getSecretKey());
        properties.put(PropertyKeyConst.NAMESRV_ADDR, aliYunMQConfiguration.getNamesrv_addr());
        properties.put(PropertyKeyConst.SendMsgTimeoutMillis, mqProperties.getSendMsgTimeout());
        Producer producer = ONSFactory.createProducer(properties);
        producer.start();

        if (ObjectUtil.isNull(producer)) {
            throw new BaseRunTimeException("获取Producer异常");
        }
        mqAdminMap.put(groupId, producer);
        return producer;
    }

    private Object process() {
        MQParameter args = this.coordinator.process();
        if (StrUtil.containsAnyIgnoreCase(this.mq_class.getName(), MQ_PRODUCER_KEY)) {
           return doSend(args.getSendArgs());
        }
        if (StrUtil.containsAnyIgnoreCase(this.mq_class.getName(), MQ_CONSUMER_KEY)) {
            return doConsume(args.getConsumeArgs());
        }
        return false;
    }

    private Object doConsume(MQParameter.ConsumeArgs args) {
        if (StrUtil.containsAnyIgnoreCase(mq_class.getName(), MQ_ORDER_KEY)) {
            OrderConsumer consumer = (OrderConsumer) this.mqAdminMap.get(this.groupId);
            consumer.subscribe(args.getTopic(), args.getTag(), args.getOrderListener());
            return true;
        }

        Consumer consumer = (Consumer)  this.mqAdminMap.get(this.groupId);
        consumer.subscribe(args.getTopic(), args.getTag(), args.getMessageListener());
        return true;
    }

    private Object doSend(MQParameter.SendArgs args){
        if (StrUtil.containsAnyIgnoreCase(mq_class.getName(), MQ_ORDER_KEY)) {
            OrderProducer producer = (OrderProducer) this.mqAdminMap.get(this.groupId);
            SendResult result = producer.send(args.getMessage(), args.getShardingKey());
            return result;
        }
        Producer producer = (Producer) this.mqAdminMap.get(this.groupId);
        SendResult result = producer.send(args.getMessage());
        return result;
    }

}
